import com.alibaba.fastjson.JSON;
import com.tc.zookeeper.curator.EnhancedCuratorFramework;
import com.tc.zookeeper.curator.EnhancedCuratorFrameworkFactory;
import com.tc.zookeeper.test.ExampleClientThatLocks;
import com.tc.zookeeper.test.FakeLimitedResource;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

//注意 curator 4 只支持 zookeeper 3.5 以上，否者创建节点失败
public class testZk {

    //private static final String CONNECT_ADDR = "127.0.0.1:2181";
    private static final String CONNECT_ADDR = "10.198.195.147:2181,10.198.195.148:2181,10.198.195.149:2181";
    private static final int SESSION_TIMEOUT = 5000;

    @Test
    public void demo0() throws Exception { //节点创建和删除
        //重试策略，初试时间1秒，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        //通过工厂创建Curator
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        //开启连接
        curator.start();
        curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c321321", "c1内容".getBytes()); //每个结点的最大数据量有一个上限是1M  默认最大上限

        //curator.delete().guaranteed().deletingChildrenIfNeeded().forPath("/default/config/gradle-demo");
        curator.close();
    }

    @Test
    public void demo() throws Exception { //节点创建和删除
        //重试策略，初试时间1秒，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        //通过工厂创建Curator
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        //开启连接
        curator.start();
        ExecutorService executor = Executors.newCachedThreadPool();
        /**创建节点，creatingParentsIfNeeded()方法的意思是如果父节点不存在，则在创建节点的同时创建父节点；
         * withMode()方法指定创建的节点类型，跟原生的Zookeeper API一样，不设置默认为PERSISTENT类型。
         * */
        curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
                .inBackground((framework, event) -> { //添加回调
                    System.out.println("---------Code：" + event.getResultCode());
                    System.out.println("---------Type：" + event.getType());
                    System.out.println("----------Path：" + event.getPath());
                }, executor).forPath("/super/c1", "c1内容".getBytes()); //每个结点的最大数据量有一个上限是1M  默认最大上限
        Thread.sleep(5000); //为了能够看到回调信息
        Stat stat = curator.checkExists().forPath("/example/basic"); //判断指定节点是否存在
        System.out.println(stat);
        String data = new String(curator.getData().forPath("/super/c1")); //获取节点数据
        System.out.println(data);
        curator.setData().forPath("/super/c1", "c1新内容".getBytes()); //更新节点数据
        data = new String(curator.getData().forPath("/super/c1"));
        System.out.println(data);
        List<String> children = curator.getChildren().forPath("/super"); //获取子节点
        for (String child : children) {
            System.out.println(child);
        }
        //放心的删除节点，deletingChildrenIfNeeded()方法表示如果存在子节点的话，同时删除子节点
        curator.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
        curator.close();
    }

    @Test
    public void demo2() throws Exception {//单节点监控
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        curator.start();
        //最后一个参数表示是否进行压缩
        NodeCache cache = new NodeCache(curator, "/super1", false);
        cache.start(true);
        //只会监听节点的创建和修改，删除不会监听
        cache.getListenable().addListener(() -> {
            System.out.println("路径：" + cache.getCurrentData().getPath());
            System.out.println("数据：" + new String(cache.getCurrentData().getData()));
            System.out.println("状态：" + cache.getCurrentData().getStat());
        });

        curator.create().forPath("/super1", "1234".getBytes());
        Thread.sleep(1000);
        curator.setData().forPath("/super1", "5678".getBytes());
        Thread.sleep(1000);
        curator.delete().forPath("/super1");
        Thread.sleep(5000);
        curator.close();
    }

    @Test
    public void demo3() throws Exception {//多节点监控
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        curator.start();
        //第三个参数表示是否接收节点数据内容
        PathChildrenCache childrenCache = new PathChildrenCache(curator, "/super", true);
        /**
         * 如果不填写这个参数，则无法监听到子节点的数据更新
         * StartMode：初始化方式
         * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
         * NORMAL：异步初始化
         * BUILD_INITIAL_CACHE：同步初始化
         *
         */
        childrenCache.start(PathChildrenCache.StartMode.NORMAL);

        childrenCache.getListenable().addListener((framework, event) -> {
            switch (event.getType()) {
                case CHILD_ADDED:
                    System.out.println("CHILD_ADDED，类型：" + event.getType() + "，路径：" + event.getData().getPath() + "，数据：" +
                            new String(event.getData().getData()) + "，状态：" + event.getData().getStat());
                    break;
                case CHILD_UPDATED:
                    System.out.println("CHILD_UPDATED，类型：" + event.getType() + "，路径：" + event.getData().getPath() + "，数据：" +
                            new String(event.getData().getData()) + "，状态：" + event.getData().getStat());
                    break;
                case CHILD_REMOVED:
                    System.out.println("CHILD_REMOVED，类型：" + event.getType() + "，路径：" + event.getData().getPath() + "，数据：" +
                            new String(event.getData().getData()) + "，状态：" + event.getData().getStat());
                    break;
                default:
                    break;
            }
        });
        curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1", "c1内容".getBytes());
        //经测试，不会监听到本节点的数据变更，只会监听到指定节点下子节点数据的变更
        curator.setData().forPath("/super", "456".getBytes());
        curator.setData().forPath("/super/c1", "c1新内容".getBytes());
        curator.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
        Thread.sleep(5000);
        curator.close();
    }

    private static final int QTY = 5;
    private static final int REPETITIONS = QTY * 10;
    private static final String PATH = "/examples/locks1";

    @Test
    public void demo4() throws Exception {//可重入锁
        final FakeLimitedResource resource = new FakeLimitedResource();
        ExecutorService executor = Executors.newFixedThreadPool(QTY);
        try {
            for (int i = 0; i < QTY; i++) {
                final int index = i;
                Callable<Void> task = () -> {
                    CuratorFramework curator = CuratorFrameworkFactory.newClient(CONNECT_ADDR, new RetryNTimes(3, 1000));
                    curator.start();
                    try {
                        final ExampleClientThatLocks example = new ExampleClientThatLocks(curator, PATH, resource, "Client " + index);
                        for (int j = 0; j < 1; j++) {
                            example.doWork(10, TimeUnit.SECONDS);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        CloseableUtils.closeQuietly(curator);
                    }
                    return null;
                };
                executor.submit(task);
            }
            executor.shutdown();
            executor.awaitTermination(10, TimeUnit.MINUTES);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void demo5() throws Exception {//双栅栏
        CuratorFramework curator = CuratorFrameworkFactory.newClient(CONNECT_ADDR, new RetryNTimes(3, 1000));
        curator.start();
        ExecutorService executor = Executors.newFixedThreadPool(5);
        for(int i=0; i<5; i++) {
            final DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(curator, "/example/barrier", 5);
            final int index = i;
            Callable<Void> task = () -> {
                Thread.sleep((long) (3000 * Math.random()));
                System.out.println("Client#" + index + " enter");
                barrier.enter();
                System.out.println("Client#" + index + "begin");
                Thread.sleep((long) (3000 * Math.random()));
                barrier.leave();
                System.out.println("Client#" + index + "left");
                return null;
            };
            executor.submit(task);
        }
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.MINUTES);
        curator.close();
    }


    @Test
    public void demo7() throws Exception {//自定义curator

        System.setProperty("VIP_CFGCENTER_ZK_CONNECTION",CONNECT_ADDR);

        String targetConnectString = EnhancedCuratorFrameworkFactory.getTargetConnectString(CONNECT_ADDR);

        EnhancedCuratorFramework curator  = EnhancedCuratorFrameworkFactory.getStartedClientByConnectString(targetConnectString);

        curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c41", "c1内容".getBytes()); //每个结点的最大数据量有一个上限是1M  默认最大上限

        curator.close();

    }


    @Test
    public void demo8() throws Exception { //节点创建和删除
        //重试策略，初试时间1秒，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        //通过工厂创建Curator
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        //开启连接
        curator.start();
        ExecutorService executor = Executors.newCachedThreadPool();
        /**创建节点，creatingParentsIfNeeded()方法的意思是如果父节点不存在，则在创建节点的同时创建父节点；
         * withMode()方法指定创建的节点类型，跟原生的Zookeeper API一样，不设置默认为PERSISTENT类型。
         * */
        curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
                .inBackground((framework, event) -> { //添加回调
                    System.out.println("---------Code：" + event.getResultCode());
                    System.out.println("---------Type：" + event.getType());
                    System.out.println("----------Path：" + event.getPath());
                }, executor).forPath("/default/config/gradle-demo", "c1内容".getBytes()); //每个结点的最大数据量有一个上限是1M  默认最大上限
        Thread.sleep(5000); //为了能够看到回调信息
        Stat stat = curator.checkExists().forPath("/default/config/gradle-demo"); //判断指定节点是否存在
        System.out.println(stat);
        String data = new String(curator.getData().forPath("/default/config/gradle-demo")); //获取节点数据
        System.out.println(data);
        curator.setData().forPath("/default/config/gradle-demo", "{\"zkp\" : \" hello 212121 adsad\"  }".getBytes()); //更新节点数据
        data = new String(curator.getData().forPath("/default/config/gradle-demo"));
        System.out.println(data);
        curator.close();
    }


    @Test
    public void demo9() throws Exception {
        //重试策略，初试时间1秒，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        //通过工厂创建Curator
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        //开启连接
        curator.start();
        String data = new String(curator.getData().forPath("/default/vms/queue")); //获取节点数据
        List<String> strings = curator.getChildren().forPath("/default/vms/queue");
        System.out.println(JSON.toJSONString(strings));

        curator.close();
    }


}




